import asyncio
from asyncio import Queue


async def producer(name, queue: Queue, max_items):
    """异步生产者"""
    for i in range(max_items):
        item = f"{name}-{i}"
        print(f"生产者 {name} 生产了 {item}")
        await queue.put(item)
        await asyncio.sleep(1)  # 模拟异步生产


async def consumer(name, queue: Queue):
    """异步消费者"""
    while True:
        item = await queue.get()
        print(f"消费者 {name} 消费了 {item}")
        queue.task_done()
        await asyncio.sleep(1)  # 模拟异步消费


async def run():
    queue: Queue = asyncio.Queue(maxsize=3)  # 限制队列大小实现背压
    producers = [
        asyncio.create_task(producer(f"P-{i}", queue, 5))
        for i in range(3)
    ]
    consumers = [
        asyncio.create_task(consumer(f"C-{i}", queue))
        for i in range(2)
    ]

    await asyncio.gather(*producers)  # 等待所有生产者完成
    await queue.join()  # 等待队列清空
    for c in consumers:  # 取消消费者任务
        c.cancel()


if __name__ == '__main__':
    asyncio.run(run())
